package com.gin.monitor.agg.monitor;

import com.gin.monitor.agg.connector.kafka.FlinkKafkaUtils;
import com.gin.monitor.agg.connector.mysql.AbstractMysqlSource;
import com.gin.monitor.agg.connector.mysql.AbstractMysqlSink;
import com.gin.monitor.agg.utils.GlobalConstants;
import com.gin.monitor.agg.utils.JobEnvUtils;
import com.gin.monitor.agg.vo.MonitorInfo;
import com.gin.monitor.agg.vo.OutOfLimitSpeedInfo;
import com.gin.monitor.agg.vo.TrafficInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;


/**
 * 实时车辆超速监控
 * cd /opt/software/kafka_2.11-2.0.0/bin
 * bin/kafka-server-start.sh -daemon config/server.properties
 * ./kafka-console-consumer.sh --bootstrap-server node01:9092 --topic t_traffic_monitor --group group01 --property print.key=true --property print.value=true --property key.separator=,
 * 测试生产
 * ./kafka-console-producer.sh --broker-list node01:9092 --topic t_traffic_monitor
 * 1592675767000,0002,75312,京L22188,80.1,27,31
 * 1592675768000,0002,75312,京L22188,88.1,27,31
 * 1592675769000,0004,75312,京L22188,300.6,27,31
 * 1592675709000,0004,75312,京L22188,300.6,27,31
 * 1592675789000,0004,75312,京L22188,300.6,27,31
 * 1592675609000,0004,75312,京L22188,300.6,27,31
 * 1592675799000,0004,75312,京L22188,300.6,27,31
 *
 * @author gin
 * @date 2021/4/30
 */
public class OutOfSpeedMonitorAnalysis {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = JobEnvUtils.initEnv(args);

        //streamKafka 海量的数据流，不可以存入广播状态流中
        //streamMysql 从Mysql数据库中读取的卡口限速信息，特点：数据量少，更新不频繁

        //默认最近开始读取
        //DataStreamSource<TrafficInfo> streamKafka = env.addSource(FlinkKafkaUtils.getKafkaConsumer());
        //从第一行开始读取. 测试用!
        SingleOutputStreamOperator<TrafficInfo> streamKafka = env
                .addSource(FlinkKafkaUtils.getKafkaConsumer().setStartFromEarliest());

        SingleOutputStreamOperator<MonitorInfo> streamMysql = env
                .addSource(new AbstractMysqlSource<MonitorInfo>() {
                })
                .returns(TypeInformation.of(new TypeHint<MonitorInfo>() {
                }));

        //streamMysql.print();

        BroadcastStream<MonitorInfo> broadcastStream = streamMysql.broadcast(GlobalConstants.MONITOR_STATE_DESCRIPTOR);

        SingleOutputStreamOperator<OutOfLimitSpeedInfo> overSpeedStream = streamKafka
                .connect(broadcastStream)
                .process(new BroadcastProcessFunction<TrafficInfo, MonitorInfo, OutOfLimitSpeedInfo>() {
                    @Override
                    public void processElement(TrafficInfo value, ReadOnlyContext ctx, Collector<OutOfLimitSpeedInfo> out) throws Exception {
                        //先从状态中得到当前卡口的限速信息
                        MonitorInfo info = ctx.getBroadcastState(GlobalConstants.MONITOR_STATE_DESCRIPTOR).get(value.monitorId);
                        //表示当前这里车，经过的卡口是有限速的
                        if (info != null) {
                            int limitSpeed = info.limitSpeed;
                            double realSpeed = value.speed;
                            //当前车辆超速通过卡口
                            //超速 1.1 倍才会收集
                            if (limitSpeed * 1.1 < realSpeed) {
                                out.collect(new OutOfLimitSpeedInfo(value.car, value.monitorId, value.roadId, realSpeed, limitSpeed, value.actionTime));
                            }
                        }
                    }

                    @Override
                    public void processBroadcastElement(MonitorInfo value, Context ctx, Collector<OutOfLimitSpeedInfo> out) throws Exception {
                        //把广播流中的数据保存到状态中
                        ctx.getBroadcastState(GlobalConstants.MONITOR_STATE_DESCRIPTOR).put(value.monitorId, value);
                    }
                });

        overSpeedStream.print();

        overSpeedStream.addSink(new AbstractMysqlSink<OutOfLimitSpeedInfo>(){});

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
